package com.geng.provider;

import com.alibaba.fastjson.JSON;
import com.geng.dto.MyRequest;
import com.geng.dto.MyResponse;
import com.geng.dto.ProviderService;
import com.geng.zkORdc.IRegisterCenter4Provider;
import com.geng.zkORdc.RegisterCenter;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.StringUtils;

import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 接收到信息，将业务处理抽离出IO，用线程池来完成
 * 1、在这里处理接收到的信息
 * 2、去调用具体实现类
 * 3、封装response从channel写回去
 */

public class WorkerThreak implements Runnable {
    //private static List<Filter> filterList = new ArrayList<>();

    //使用静态代码快来尽早捕获
   /* static {
        ServiceLoader<Filter> serviceLoader = ServiceLoader.load(Filter.class);
        Iterator<Filter> iterator = serviceLoader.iterator();
        while (iterator.hasNext()) {
            filterList.add(iterator.next());
        }
        filterList.add(new DefaultServerFilter());
        Collections.sort(filterList, new Comparator<Filter>() {
            @Override
            public int compare(Filter o1, Filter o2) {
                if (o1.getPriority() == o2.getPriority()) {
                    return 0;
                }
                return o1.getPriority() < o2.getPriority() ? -1 : 1;
            }
        });
    }*/

   //直连时候无法获得provider端的线程，则初始化一个
   private static final int DNS_THREAD = 10;
    /**
     * 服务端限流
     */
    private static final Map<String, Semaphore> serviceKeySemaphoreMap = new ConcurrentHashMap<>();
    private ChannelHandlerContext ctx;
    private MyRequest request;

    public WorkerThreak(ChannelHandlerContext ctx, MyRequest request) {
        this.ctx = ctx;
        this.request = request;
    }

    @Override
    public void run() {
        //从服务调用对象里获取服务提供者信息
        ProviderService metaDataModel = request.getProviderService();
        long consumeTimeOut = request.getInvokeTimeout();
        final String methodName = request.getInvokedMethodName();

        //根据方法名称定位到具体某一个服务提供者
        String serviceKey = metaDataModel.getServiceItf().getName();
        //获取限流工具类
        //如果采用直连，则初始化一个值
        int workerThread = metaDataModel.getWorkerThreads() == 0 ? DNS_THREAD : metaDataModel.getWorkerThreads();
        Semaphore semaphore = serviceKeySemaphoreMap.get(serviceKey);
        if (semaphore == null) {
            synchronized (serviceKeySemaphoreMap) {
                semaphore = serviceKeySemaphoreMap.get(serviceKey);
                if (semaphore == null) {
                    semaphore = new Semaphore(workerThread);
                    serviceKeySemaphoreMap.put(serviceKey, semaphore);
                }
            }
        }

        MyResponse response = new MyResponse();
        //获取注册中心服务
        IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
        List<ProviderService> localProviderCaches = registerCenter4Provider.getProviderServiceMap().get(serviceKey);

        Object result = null;
        boolean acquire = false;

        try {
            // 从localProviderCaches中筛选出与当前客户端要调用的方法和端口一致对应的ProviderService
            ProviderService localProviderCache = null;
            for(ProviderService providerService : localProviderCaches){
                if(StringUtils.equals(providerService.getServiceMethod().getName(), methodName)
                        && providerService.getServerPort() == request.getProviderService().getServerPort()){
                    localProviderCache = providerService;
                    break;
                }
            }

            Object serviceObject = localProviderCache.getServiceObject();
            //System.out.println(JSON.toJSONString(localProviderCache) + localProviderCache.getServiceMethod().getName());
            // 利用反射发起服务调用，由于之前已经对method对象进行了缓存，所以这里不需要重新查找
            Method method = localProviderCache.getServiceMethod();
            //利用semaphore实现限流
            acquire = semaphore.tryAcquire(consumeTimeOut, TimeUnit.MILLISECONDS);
            if (acquire) {
                result = method.invoke(serviceObject, request.getArgs());
                //System.out.println("---------------"+result);
            }
        } catch (Exception e) {
            System.out.println(JSON.toJSONString(localProviderCaches) + "  " + methodName+" "+e.getMessage());
            result = e;
        } finally {
            if (acquire) {
                semaphore.release();
            }
        }

        // 根据服务调用结果组装调用返回对象
        response.setInvokeTimeout(consumeTimeOut);
        response.setUniqueKey(request.getUniqueKey());
        response.setResult(result);
        ctx.writeAndFlush(response);
    }
}
